package cn.springboot.linked;
/**
 * @author liufl E-mail:Huberier@allpyra.com
 * @version 创建时间：2017年9月27日 下午6:11:57
 * 类说明
 */
import java.io.IOException;  
import java.util.StringTokenizer;  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.mapreduce.Mapper;  
import org.apache.hadoop.mapreduce.Reducer;  
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.input.FileSplit;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
import org.apache.hadoop.util.GenericOptionsParser;  
  
public class InvertedIndex  
{  
      
    public static class IndexMapper extends Mapper<Object, Text, Text, Text>  
    {  
        private Text keyInfo=new Text();  
        private Text valueInfo=new Text();  
        private FileSplit split;  
        @Override  
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException  
        {  
            //获取<key value>对所属的FileSplit对象  
            split=(FileSplit) context.getInputSplit();  
            StringTokenizer iter=new StringTokenizer(value.toString());  
            while(iter.hasMoreTokens())  
            {  
                keyInfo.set(iter.nextToken().toString()+":"+split.getPath().toString());  
                valueInfo.set("1");  
                context.write(keyInfo, valueInfo);  
            }  
        }  
    }   
    
    public static class IndexReducer extends Reducer<Text, Text, Text, Text>  
    {  
	      public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException  
	      {  
	          String valueList=new String(); //不能写到外面去  
	          for(Text val:values)  
	          {  
	              valueList+=val.toString()+";";  
	          }  
	          context.write(key, new Text(valueList));  
	      }  
    }
      
    public static class IndexCombiner extends Reducer<Text, Text, Text, Text>  
    {  
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException  
        {  
            int sumCount=0;  
            for(Text val:values)  
            {  
                sumCount+=Integer.parseInt(val.toString());  
            }  
            String keyInfo=key.toString().split(":")[0];  
            String[] temp=key.toString().split("/");  
            String valueInfo=temp[temp.length-1]+":"+sumCount;  
            context.write(new Text(keyInfo), new Text(valueInfo));  
        }  
    }   
      
    public static void main(String[] args) throws Exception  
    {  
        Configuration conf=new Configuration();  
        String[] otherArgs=new GenericOptionsParser(conf, args).getRemainingArgs();  
        if(otherArgs.length!=2)  
        {  
            System.out.println("Usage: InvertedIndex <in> <out>");  
            System.exit(2);  
        }  
        Path inputPath=new Path(args[0]);  
        Path outputPath=new Path(args[1]);  
        outputPath.getFileSystem(conf).delete(outputPath, true);  
          
        //The constructor Job(Configuration conf, String jobName) is deprecated  
        //Job job=new Job(conf, "Inverted Index");  
        Job job=Job.getInstance(conf, "Inverted Index");  
        job.setJarByClass(InvertedIndex.class);  
          
        job.setMapperClass(IndexMapper.class);  
        job.setMapOutputKeyClass(Text.class);  
        job.setMapOutputValueClass(Text.class);  
          
        job.setCombinerClass(IndexCombiner.class);  
          
        job.setReducerClass(IndexReducer.class);          
        job.setOutputKeyClass(Text.class);  
        job.setOutputValueClass(Text.class);  
          
        FileInputFormat.addInputPath(job, inputPath);  
        FileOutputFormat.setOutputPath(job, outputPath);  
          
        System.exit(job.waitForCompletion(true)? 0:1);  
    }  
}

